package com.swsc.ai.tasks;

import com.swsc.ai.config.SparkSqlConf;
import com.swsc.ai.enums.TagTypeEnum;
import org.apache.spark.sql.*;

import java.util.ArrayList;

/**
 * @author QF
 * @date 2023/10/23 11:25
 * @describe 用户偏好标签计算
 */
public class UserFundPreferenceTask extends UserPreferenceTemplateTask {

    @Override
    public void createAction(SparkSession session, String actionPath, String wideTablePath) throws RuntimeException {
        Dataset<Row> test = null;
        try {
            test = SparkSqlConf.getDataByCVS(session, "user_id STRING, behive STRING", actionPath);
        }catch (Exception e){
            throw new RuntimeException("读取今日购买行为异常！");
        }
        test.select("user_id", "behive").createOrReplaceTempView("myaction");
        Dataset<Row> sql = session.sql("select user_id,size(split(behive, '\t')),split(behive, '\t') AS actionList from myaction");
        // 使用flatMap函数进行扁平化操作
        Dataset<Row> flatDataset = sql.select(functions.explode(new Column("actionList")), new Column("user_id"))
                .selectExpr("col AS action", "user_id");
        flatDataset.createOrReplaceTempView("user_action");
        Dataset<Row> sql1 = session.sql("select user_id, split(action,':')[0] as tag," +
                "(CASE\n" +
                "    WHEN split(action,':')[1] = 2 THEN -0.2\n" +
                "    ELSE 1\n" +
                " END) * split(action,':')[2] * split(action,':')[3] as amount, split(action,':')[4] buy_date,'astockcode' tag_type from user_action");
        sql1.createOrReplaceTempView("user_action_data");
        //创建宽表视图
        Dataset<Row> wideTable = null;
        try {
            wideTable = SparkSqlConf.getDataByCVS(session, "innercode STRING, main_code STRING, companycode STRING, typecode STRING, fundnatureid STRING, investmenttypecode STRING, investstylecode STRING, fundtypecode STRING, floattypecode STRING, foundedsize STRING, investadvisorcode STRING, datacode STRING, RiskLevel STRING, personalcode STRING, gendercode STRING, nationalitycode STRING,  age STRING, educationcode STRING, experiencetime STRING, returntypeavg STRING, returntyperank STRING, monrettypeavg STRING, monrettyperank STRING, totalaumtypeavg STRING, totalaumrank STRING, avgaumtypeavg STRING, avgaumtyperank STRING, enddate STRING, prod_name STRING", wideTablePath);
        }catch (Exception e){
            throw new RuntimeException("读取宽表数据异常！");
        }
        wideTable.createOrReplaceTempView("wideTable");
        ArrayList<Dataset<Row>> datasets = new ArrayList<>();
        //各维度统计
        TagTypeEnum[] values = TagTypeEnum.values();
        for (TagTypeEnum value : values) {
            if (1 == value.getTypeCode()){
                String name = value.name();
                Dataset<Row> dataset = session.sql("select br.user_id AS user_id,wt." +value.name()+ " AS tag,'" + name + "' AS tag_type,br.amount AS amount,br.buy_date AS buy_date from user_action_data br LEFT JOIN wideTable wt where br.tag = wt.innercode");
                datasets.add(dataset);
            }
            if (3 == value.getTypeCode()){
                //基金非宽表字段
                Dataset<Row> rowDataset = session.sql("select br.user_id AS user_id, 1 AS tag,'" + value.name() + "' AS tag_type,br.amount AS amount,br.buy_date AS buy_date from user_action_data br");
                datasets.add(rowDataset);
            }
        }
        Dataset<Row> union = datasets.get(0)
                .union(datasets.get(1))
                .union(datasets.get(2))
                .union(datasets.get(3))
                .union(datasets.get(4))
                .union(datasets.get(5))
                .union(datasets.get(6))
                .union(datasets.get(7))
                .union(datasets.get(8))
                .union(datasets.get(9));
        union.createOrReplaceTempView("user_action_data");

    }
}
